package com.semidata.trp.frequency;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.aredis.net.ConnectionStatus;

import com.semidata.rtd.core.datasource.DataSourceCompleteHandler;
import com.semidata.rtd.core.datasource.DataSourceErrorHandler;
import com.semidata.rtd.core.datasource.redis.RedisDataSource;
import com.semidata.rtd.core.log.Logger;
import com.semidata.rtd.core.redis.RedisClient;
import com.semidata.rtd.core.redis.RedisResponseHandler;

public class Frequency {
	
	private RedisDataSource datasource = null;
	private FrequencyHandler writeHandler = null;
	private FrequencyHandler readFrequencyHandler = null;
	private FrequencyHandler readTimestampHandler = null;

	public Frequency(RedisDataSource datasource) {
		this.datasource = datasource;

	}

	public void setWriteHandler(FrequencyHandler writeHandler) {
		this.writeHandler = writeHandler;
	}

	public void setReadFrequencyHandler(FrequencyHandler readFrequencyHandler) {
		this.readFrequencyHandler = readFrequencyHandler;
	}

	public void setReadTimestampHandler(FrequencyHandler readTimestampHandler) {
		this.readTimestampHandler = readTimestampHandler;
	}
	public void writeExpire(String userId, int expire, int timeout) {
		RedisClient client = this.datasource.getClient();
		client.expire(userId, expire, timeout, new RedisResponseHandler() {
			
			@Override
			public void handle(Object result, Exception cause) {

			}
		});
	}
	public void readFrequency(final String userId) {
		readFrequency(userId, 50);
	}

	public void readFrequency(final String userId, final int timeout) {
		final RedisClient client = this.datasource.getClient();
		client.getAll(userId, timeout, new RedisResponseHandler() {
			
			@Override
			public void handle(Object result, Exception cause) {
				if (result == null | cause != null) {
					readFrequencyHandler.handler(null);
				} else {
					List<Object> results = (List<Object>)result;
					Map<String,Object> finalResult = new HashMap<String,Object>();
					String value = "";
					String key = "";
					for (int i = 1; i <= results.size(); i++) {
						if (i % 2 == 0) {
							value = new String((byte [])results.get(i - 1));
							finalResult.put(key, value);
						} else {
							key = new String((byte [])results.get(i - 1));
						}
					}
					Map<String,Integer> frequency = processResults(finalResult, userId);
					readFrequencyHandler.handler(frequency);
				}
				
			}
		});

	}

	public void readTimestamp(final String userId, final String mid, final int timeout) {
		final RedisClient client = this.datasource.getClient();
		client.get(userId, mid, timeout, new RedisResponseHandler() {
			
			@Override
			public void handle(Object result, Exception cause) {
				// TODO Auto-generated method stub
				List<String> results = new ArrayList<String>();
				if (result == null || cause != null) {
					
					readTimestampHandler.handler(results);
				} else {
					String times = new String((byte [])result);
					
					if (times.length() != 0) {
						String[] ts = times.split(" ");
						for (int i = 0; i < ts.length; i++)
							results.add(ts[i]);
					}
					readTimestampHandler.handler(results);
				}
			}
		});
	}

	public void write(final String userId, final String mid,
	        final String timestamp) {
		write(userId, mid, timestamp, 0);
	}

	public void write(final String userId, final String mid,
	        final String timestamp, final int timeout) {
		
		setReadTimestampHandler(new FrequencyHandler() {

			public void handler(Object o) {
				List<String> results = (List<String>) o;
				long time = getCurrentTime(System.currentTimeMillis());
				for (int i = 0; i < results.size(); i++) {
					String value = results.get(i);
					if (value.length() == 0) {
						results.remove(i);
						continue;
					}
					if (time > Long.parseLong(value)) {
						results.remove(i);
					}
				}
				// System.out.println("start write in frequency");
				StringBuffer times = new StringBuffer();
				RedisClient client = datasource.getClient();
				for (String m : results) {
					if (m.length() == 0)
						continue;
					times.append(m);
					times.append(" ");
				}
				Long secTime = Long.parseLong(timestamp) / 1000;
				times.append(secTime);
				client.set(userId, mid, times.toString(), timeout, new RedisResponseHandler() {
					
					@Override
					public void handle(Object result, Exception cause) {
						// TODO Auto-generated method stub
						writeHandler.handler(null);
					}
				});

			}
		});
		readTimestamp(userId, mid, timeout);
	}

	public Map<String, Integer> processResults(Map<String, Object> results,
	        String userId) {

		Map<String, Integer> data = new HashMap<String, Integer>();
		if (results == null)
			return data;
		String key = "";
		String value = "";
		for (Map.Entry<String, Object> entry : results.entrySet()) {
			key = entry.getKey();
			value = (String) entry.getValue();
			Set<Long> times = new HashSet<Long>();
			long currentTime = this.getCurrentTime(System.currentTimeMillis());
			String[] time = value.split(" ");
			for (String t : time) {
				if (t.length() == 0)
					continue;
				long timestamp = Long.parseLong(t);
				if (currentTime <= timestamp)
					times.add(timestamp);
			}
			data.put(key, times.size());
		}
		return data;
	}

	public long getCurrentTime(long timestamp) {
		Calendar ca = Calendar.getInstance();
		ca.setTimeInMillis(timestamp);
		ca.add(Calendar.HOUR, -24);
		return ca.getTimeInMillis() / 1000;
	}
	public static void main(String[] args) {
		String []hosts = {"d189.mzhen.cn:6381"};
		final RedisDataSource redis = new RedisDataSource(hosts,8,3,3000);
		Frequency fre = new Frequency(redis);
		fre.setReadFrequencyHandler(new FrequencyHandler() {

			@SuppressWarnings("unchecked")
			@Override
			public void handler(Object o) {
				// TODO Auto-generated method stub
				Map<String,Integer> results = null;
				System.out.println("lalalalalal");
				if (o == null) {
					results = new HashMap<String, Integer>();
				} else {
					results = (Map<String, Integer>) o;
				}
				System.out.println(results.size());
				for (Map.Entry<String, Integer> m : results.entrySet())
					System.out.println("key:" + m.getKey() + " value : "
					        + m.getValue());
				//redis.getClient().stop();
			}
		});
		fre.readFrequency("5yIged", 100);
		Frequency fre1 = new Frequency(redis);
		fre1.setWriteHandler(new FrequencyHandler() {
			
			@Override
			public void handler(Object o) {
				// TODO Auto-generated method stub
				//System.out.println("increment a frequency finished: userid : " + userId + " materialid : " + mid);
				System.out.println( "Write frequency finished");
			}
	        		
		});
	
		fre1.write("mydream", "234", System.currentTimeMillis()+"", 100);
		
	}
}
